package com.prj.ufdm.demo.tutorial.thread;

import java.util.PriorityQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 使用 Condition 解决生产者、消费者问题
 * @author Administrator
 *
 */
public class TestConditionProsume {
	
    private int queueSize = 10;  
    private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
    private Lock lock = new ReentrantLock();  
    private Condition produceCondition = lock.newCondition();  
    private Condition consumeCondition = lock.newCondition(); 
    
	public static void main(String[] args) throws Exception {
		TestConditionProsume test = new TestConditionProsume();  
		TCProducer p1 = test.new TCProducer(); 
		TCProducer p2 = test.new TCProducer(); 
		TCProducer p3 = test.new TCProducer(); 
        TCConsumer c1 = test.new TCConsumer();
        TCConsumer c2 = test.new TCConsumer();   
        TCConsumer c3 = test.new TCConsumer(); 
        ExecutorService threadPool = Executors.newCachedThreadPool();
        threadPool.execute(p1);
        threadPool.execute(p2);
        threadPool.execute(p3);
        threadPool.execute(c1);
        threadPool.execute(c2);
        threadPool.execute(c3);
	}

    class TCProducer extends Thread{              
        @Override  
        public void run() {  
            while(true){  
                lock.lock();  
                try {
                    while(queue.size() == queueSize){
                        System.out.println(Thread.currentThread().getName()+"生产者：队列满，等待有空余空间");  
                        produceCondition.await();  
                    }  
                    Thread.sleep(2000);
                    queue.offer(1);
                    consumeCondition.signal();  
                    System.out.println(Thread.currentThread().getName()+"生产者：生产一个元素，队列剩余空间："+(queueSize-queue.size()));  
                } 
                catch (Exception e) {  
                	e.printStackTrace();
                }  
                finally{  
                    lock.unlock();  
                }  
            }  
        }  
    }
    
	class TCConsumer extends Thread {
        @Override  
        public void run() {
            while(true){
                lock.lock();
                try {
                    while(queue.size() == 0){  
                        System.out.println(Thread.currentThread().getName()+"消费者：队列空，等待数据");  
                        consumeCondition.await();  
                    }  
                    Thread.sleep(3000);
                    queue.poll();  //每次移走队首元素  
                    produceCondition.signal();  
                    System.out.println(Thread.currentThread().getName()+"消费者：消费一个元素，队列剩余"+queue.size()+"个元素");  
                } 
                catch (Exception e) {  
                	e.printStackTrace();
                }  
                finally{  
                    lock.unlock();  
                }  
            }   
        }  
    }          
  
}
